package mapReduce;

import java.io.IOException;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;

import util.Utils;

public class TopTenPhotosMR {

	public static class Mapper extends
			TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {
		
		protected void map(ImmutableBytesWritable key, Result result,
				Context context) throws IOException, InterruptedException {

			String text = Bytes.toString(result
					.getRow());

			ImmutableBytesWritable immutableKeyPhoto = new ImmutableBytesWritable(
					Bytes.toBytes(text.substring(0, Utils.SIZE_UUID*2)));
			
			ImmutableBytesWritable immutableKeyComment = new ImmutableBytesWritable(
					Bytes.toBytes(text.substring(0, text.length() - (Utils.SIZE_UUID*2) )));;
					
			context.write(immutableKeyPhoto, immutableKeyComment);
		}

	}

	public static class Reducer
			extends
			TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, NullWritable> {

		private static Map<String, Integer> map = new HashMap<String, Integer>();

		protected void reduce(ImmutableBytesWritable key,
				Iterable<ImmutableBytesWritable> it, Context context)
				throws IOException, InterruptedException {

			String word = Bytes.toString(key.get());
			int count = 0;
			for (ImmutableBytesWritable immutableBytesWritable : it) {
				count++;
			}
			map.put(word, count);
		}

		public static String printTopTenPhotos() {
			List<Map.Entry<String, Integer>> list = new LinkedList<Map.Entry<String, Integer>>(
					map.entrySet());
			Collections.sort(list,
					new Comparator<Map.Entry<String, Integer>>() {
						public int compare(Entry<String, Integer> e1,
								Entry<String, Integer> e2) {
							return e2.getValue() - e1.getValue();
						}
					});
			return list.size() < 10?list.toString():list.subList(0, 10).toString();
		}

	}
}
